package com.dmp.storm.bolt.socket;

import com.dmp.storm.bolt.BaseMapBolt;
import com.dmp.storm.bolt.file.BaseFileBolt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.Socket;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 
 * @ClassName: BaseSocketBolt
 * @Description: Socket基类
 * @author suzg
 * @date 2015年11月5日 下午4:15:05
 *
 */
public abstract class BaseSocketBolt extends BaseMapBolt {

	private static final long serialVersionUID = 1L;
	private static Logger log = LoggerFactory.getLogger(BaseSocketBolt.class);
	private static Properties props = new Properties();
	private Map<String, String> redisMap;
	// 配置信息
	protected String extraKeys;// 额外需要处理的字段
	private Socket socket = null;
	private String serverSocketIp = null;
	private int serverSocketPort = 0;

	OutputStreamWriter writer = null;

	@Override
	public void prepare() {
		try {
			log.info("starting a new socket..................................");
			props.load(BaseFileBolt.class.getClassLoader().getResourceAsStream("dpi4g_info.properties"));
			serverSocketIp = props.getProperty("socket.serverIp");
			serverSocketPort = Integer.parseInt(props.getProperty("socket.serverPort"));
			log.info("serverSocketIp = " + serverSocketIp);
			log.info("serverSocketPort = " + serverSocketPort);
			redisMap = new ConcurrentHashMap<String, String>();
			writer = getWriter();
		} catch (Exception e) {
			e.printStackTrace();
			System.err.println("加载配置文件出错！");
			log.error(e.getMessage());
		}
	}

	private OutputStreamWriter getWriter() throws Exception {
		socket = new Socket(serverSocketIp, serverSocketPort);
		OutputStream out = socket.getOutputStream();
		BufferedOutputStream bufferOut = new BufferedOutputStream(out);
		writer = new OutputStreamWriter(bufferOut, "UTF-8");
		log.info("get a writer for socekt");
		return writer;
	}

	/**
	 * 主执行方法
	 */
	@Override
	public Map<String, Object> process(Map<String, Object> map) {
		try {
			String message = writeMessageByMap(map);

			sendMessage(message);
		} catch (Exception e) {
			log.error(e.getMessage());
		}
		return map;

	}

	/**
	 * 
	 * @Title: sendMessage
	 * @Description: 发送消息
	 * @author suzg
	 * @date 2015年11月18日 上午9:50:35
	 * @param message
	 * @throws Exception
	 */
	private void sendMessage(String message) throws Exception {
		writer.write(message);
		// log.info("发送消息成功！");
		writer.flush();
	}

	/**
	 * 
	 * @Title: writeMessageByMap
	 * @Description: 生成要发送的消息message
	 * @author suzg
	 * @date 2015年11月18日 上午9:47:47
	 * @param map
	 * @return
	 */
	private String writeMessageByMap(Map<String, Object> map) {

		String result = "";
		String kafkamsg = (String) map.get("result");
		StringBuffer sb = new StringBuffer(kafkamsg);
		if (!kafkamsg.endsWith("|")) {
			sb.append("|");
		}
		try {
			// 处理额外的字段
			dealWithExtraKeys(sb, map);
			// stringbuffer优化：1，删除最后的“|”，2：增加换行
			sb.deleteCharAt(sb.length() - 1).append("\r\n");
			// 行内容
			result = sb.toString();

		} catch (Exception e) {
			log.error("生成socket消息失败====");
			e.printStackTrace();
		}
		return result;
	}

	public static void main(String[] args) {
		Socket socket = null;
		try {
			socket = new Socket("10.10.139.164", 30000);
			OutputStream out = socket.getOutputStream();

			socket.getInputStream();
			PrintWriter print = new PrintWriter(out);
			print.write("zhuangye,,,one,two,three \r\n");
			print.flush();
			System.out.println("发送完毕");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (socket != null) {
				try {
					socket.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 
	 * @Title: dealWithExtraKeys
	 * @Description: 处理额外的字段
	 * @author suzg
	 * @date 2015年11月6日 上午10:24:00
	 * @param sb
	 * @param map
	 * @return
	 */
	protected abstract void dealWithExtraKeys(StringBuffer sb, Map<String, Object> map) throws Exception;



	public String getExtraKeys() {
		return extraKeys;
	}

	public void setExtraKeys(String extraKeys) {
		this.extraKeys = extraKeys;
	}

}
